package com.demo.common.redis.delay;

import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.stream.Task;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * <p>
 * 延迟消息轮训 task 的实现
 * </p>
 *
 * @author molong
 * @date 2022/6/22
 */
@Slf4j
public class DelayPollTask implements Task {

    /**
     * 消息监听
     */
    private final DelayMessageListener listener;

    /**
     * 发送错误时的处理程序
     */
    //private final ErrorHandler errorHandler;

    /**
     * 在发生错误时判断是否可以取消订阅
     */
    //private final Predicate<Throwable> cancelSubscriptionOnError;

    /**
     * 读取记录的方法
     */
    private final Function<String, List<ZSetOperations.TypedTuple<DelayMessage>>> readFunction;

    /**
     * 当前任务的轮训状态
     */
    private final PollState pollState;
    private volatile boolean isInEventLoop = false;

    DelayPollTask(DelayMessageListener listener, Function<String, List<ZSetOperations.TypedTuple<DelayMessage>>> readFunction) {
        this.listener = listener;
        this.readFunction = readFunction;
        this.pollState = createPollState();
    }

    /**
     * 创建轮训状态对象
     *
     * @return PollState
     */
    private static PollState createPollState() {
        return PollState.standalone();
    }

    @Override
    public void cancel() throws DataAccessResourceFailureException {
        this.pollState.cancel();
    }

    @Override
    public State getState() {
        return pollState.getState();
    }

    @Override
    public boolean awaitStart(Duration timeout) throws InterruptedException {
        return pollState.awaitStart(timeout.toNanos(), TimeUnit.NANOSECONDS);
    }

    @Override
    public boolean isLongLived() {
        return true;
    }

    @Override
    public void run() {

        pollState.starting();

        try {

            isInEventLoop = true;
            pollState.running();
            doLoop();
        } finally {
            isInEventLoop = false;
        }
    }

    private void doLoop() {
        do {
            try {
                // allow interruption
                //休眠50毫秒
                Thread.sleep(50);
                //读取记录
                List<ZSetOperations.TypedTuple<DelayMessage>> raw = readRecords();
                //反序列化后交给监听器
                deserializeAndEmitRecords(raw);
            } catch (InterruptedException e) {
                cancel();
                Thread.currentThread().interrupt();
            } catch (RuntimeException e) {
                log.error("延迟消息拉取任务运行异常：", e);
                /*if (cancelSubscriptionOnError.test(e)) {
                    cancel();
                }

                errorHandler.handleError(e);*/
            }
        } while (pollState.isSubscriptionActive());
    }

    /**
     * 读取记录
     * @return List<ZSetOperations.TypedTuple<?>>
     */
    private List<ZSetOperations.TypedTuple<DelayMessage>> readRecords() {
        return readFunction.apply(listener.getRedisKey());
    }

    /**
     * 反序列化后交给监听器
     * @param records 记录
     */
    private void deserializeAndEmitRecords(List<ZSetOperations.TypedTuple<DelayMessage>> records) {
        for (ZSetOperations.TypedTuple<DelayMessage> typedTuple : records) {
            try {
                listener.onMessage(typedTuple.getValue());
            } catch (RuntimeException e) {
                log.error("处理异常:", e);
                //处理异常如果处理这个任务暂时不定
            /*if (cancelSubscriptionOnError.test(e)) {

                cancel();
                errorHandler.handleError(e);

                return;
            }
            errorHandler.handleError(e);*/
            }
        }
    }

    /*private V convertRecord(ZSetOperations.TypedTuple<?> record) {

        try {
            return deserializer.apply(record);
        } catch (RuntimeException e) {
            throw new ConversionFailedException(TypeDescriptor.forObject(record), targetType, record, e);
        }
    }*/

    @Override
    public boolean isActive() {
        return State.RUNNING.equals(getState()) || isInEventLoop;
    }

    /**
     * Object representing the current polling state for a particular stream subscription.
     */
    static class PollState {

        private volatile State state = State.CREATED;
        private volatile CountDownLatch awaitStart = new CountDownLatch(1);

        private PollState() {
        }

        /**
         * 创建一个状态对象
         */
        static PollState standalone() {
            return new PollState();
        }

        boolean awaitStart(long timeout, TimeUnit unit) throws InterruptedException {
            return awaitStart.await(timeout, unit);
        }

        public State getState() {
            return state;
        }

        /**
         * @return {@literal true} if the subscription is active.
         */
        boolean isSubscriptionActive() {
            return state == State.STARTING || state == State.RUNNING;
        }

        /**
         * Set the state to {@link State#STARTING}.
         */
        void starting() {
            state = State.STARTING;
        }

        /**
         * Switch the state to {@link State#RUNNING}.
         */
        void running() {

            state = State.RUNNING;

            CountDownLatch awaitStart = this.awaitStart;

            if (awaitStart.getCount() == 1) {
                awaitStart.countDown();
            }
        }

        /**
         * Set the state to {@link State#CANCELLED} and re-arm the
         * {@link #awaitStart(long, TimeUnit) await synchronizer}.
         */
        void cancel() {

            awaitStart = new CountDownLatch(1);
            state = State.CANCELLED;
        }
    }
}
